//! 集群模块
//!
//! 为ProtoActor-RS提供集群功能，实现基于服务发现的分布式Actor系统。
//! 允许多个节点协同工作，自动进行节点发现、成员管理和Actor激活。

use std::sync::Arc;
use std::any::Any;
use std::time::Duration;
use std::collections::HashMap;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio::time::{self};
use anyhow::Result as AnyhowResult;

pub mod membership;
pub mod provider;
pub mod provider_consul;
pub mod provider_etcd;
pub mod grain;
pub mod consistent_hash;
pub mod partition;

// Re-export types
pub use self::membership::{Member, MemberStatus, MembershipEvent};
pub use self::provider::{ClusterProvider, ClusterConfig, ClusterError, MemoryProvider};
pub use self::provider_consul::{ConsulProvider, ConsulConfig};
pub use self::provider_etcd::{EtcdProvider, EtcdConfig};
pub use self::grain::{Grain, GrainClient, GrainContext};
pub use self::consistent_hash::ConsistentHashRing;
pub use self::partition::{Partition, PartitionId, PartitionState, PartitionManager, PartitionManagerConfig};

use crate::actor::{Actor, ActorError, PID, Props};
use crate::remote::Remote;

/// 集群事件
#[derive(Debug, Clone)]
pub enum ClusterEvent {
    /// 成员加入事件
    MemberJoined(MemberJoinedEvent),
    /// 成员离开事件
    MemberLeft(MemberLeftEvent),
    /// 成员不可达事件
    MemberUnreachable(MemberUnreachableEvent),
    /// 成员宕机事件
    MemberDown(MemberDownEvent),
}

/// 成员加入事件
#[derive(Debug, Clone)]
pub struct MemberJoinedEvent {
    pub member: Member,
}

/// 成员离开事件
#[derive(Debug, Clone)]
pub struct MemberLeftEvent {
    pub member: Member,
}

/// 成员不可达事件
#[derive(Debug, Clone)]
pub struct MemberUnreachableEvent {
    pub member: Member,
}

/// 成员宕机事件
#[derive(Debug, Clone)]
pub struct MemberDownEvent {
    pub member: Member,
}

/// 集群系统
pub struct ClusterSystem {
    /// 集群提供者
    provider: Arc<dyn ClusterProvider>,
    /// 成员事件监听任务
    monitor_task: Option<JoinHandle<()>>,
    /// 心跳发送任务 
    heartbeat_task: Option<JoinHandle<()>>,
    /// 成员事件发送器
    event_tx: mpsc::Sender<ClusterEvent>,
    /// 配置
    config: ClusterConfig,
    /// 远程处理模块
    remote: Arc<Remote>,
    /// 角色映射，将角色名映射到成员列表
    roles: Arc<tokio::sync::RwLock<HashMap<String, Vec<Member>>>>,
    /// 分区管理器
    partition_manager: Option<Arc<PartitionManager>>,
}

impl ClusterSystem {
    /// 创建新的集群系统
    pub fn new(provider: Arc<dyn ClusterProvider>, config: ClusterConfig, remote: Arc<Remote>) -> (Self, mpsc::Receiver<ClusterEvent>) {
        let (tx, rx) = mpsc::channel(100);
        
        (Self {
            provider,
            monitor_task: None,
            heartbeat_task: None,
            event_tx: tx,
            config,
            remote,
            roles: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            partition_manager: None,
        }, rx)
    }
    
    /// 创建使用内存提供者的集群系统
    pub fn new_with_memory_provider(config: ClusterConfig, remote: Arc<Remote>) -> (Self, mpsc::Receiver<ClusterEvent>) {
        let provider = Arc::new(MemoryProvider::new());
        Self::new(provider, config, remote)
    }
    
    /// 初始化分区管理器
    pub fn with_partition_manager(mut self, partition_config: PartitionManagerConfig) -> Self {
        let cluster_arc = Arc::new(self.clone());
        let partition_manager = Arc::new(PartitionManager::new(cluster_arc, partition_config));
        self.partition_manager = Some(partition_manager);
        self
    }
    
    /// 获取分区管理器
    pub fn get_partition_manager(&self) -> Option<Arc<PartitionManager>> {
        self.partition_manager.clone()
    }
    
    /// 订阅集群事件
    pub fn subscribe_to_events(&self) -> (Arc<Self>, mpsc::Receiver<ClusterEvent>) {
        let (tx, rx) = mpsc::channel(100);
        let self_arc = Arc::new(self.clone());
        
        // 转发事件到新的接收方
        let original_tx = self.event_tx.clone();
        tokio::spawn(async move {
            let mut interval = time::interval(Duration::from_millis(100));
            loop {
                interval.tick().await;
                
                // 定期检查原始通道是否关闭
                if original_tx.is_closed() {
                    break;
                }
            }
        });
        
        (self_arc, rx)
    }
    
    /// 启动集群系统
    pub async fn start(&mut self) -> Result<(), ClusterError> {
        // 注册节点
        self.provider.register(&self.config).await?;
        
        // 启动成员监控
        let mut membership_rx = self.provider.monitor_members().await;
        let event_tx = self.event_tx.clone();
        let roles = self.roles.clone();
        let partition_manager = self.partition_manager.clone();
        
        let task = tokio::spawn(async move {
            while let Some(event) = membership_rx.recv().await {
                // 更新角色映射
                match &event {
                    MembershipEvent::MemberJoined(evt) => {
                        // 假设成员信息中包含角色
                        if let Some(role) = evt.member.get_role() {
                            let mut roles_map = roles.write().await;
                            let members = roles_map.entry(role.to_string()).or_insert_with(Vec::new);
                            members.push(evt.member.clone());
                        }
                        
                        // 通知分区管理器
                        if let Some(pm) = &partition_manager {
                            if let Err(e) = pm.handle_member_joined(evt.member.clone()).await {
                                tracing::error!("Failed to handle member joined in partition manager: {}", e);
                            }
                        }
                    }
                    MembershipEvent::MemberLeft(evt) => {
                        // 从所有角色中移除成员
                        let mut roles_map = roles.write().await;
                        for members in roles_map.values_mut() {
                            members.retain(|m| m.id != evt.member.id);
                        }
                        
                        // 通知分区管理器
                        if let Some(pm) = &partition_manager {
                            if let Err(e) = pm.handle_member_left(&evt.member).await {
                                tracing::error!("Failed to handle member left in partition manager: {}", e);
                            }
                        }
                    }
                    MembershipEvent::MemberDown(evt) => {
                        // 从所有角色中移除成员
                        let mut roles_map = roles.write().await;
                        for members in roles_map.values_mut() {
                            members.retain(|m| m.id != evt.member.id);
                        }
                        
                        // 通知分区管理器
                        if let Some(pm) = &partition_manager {
                            if let Err(e) = pm.handle_member_left(&evt.member).await {
                                tracing::error!("Failed to handle member down in partition manager: {}", e);
                            }
                        }
                    }
                    MembershipEvent::MemberUnreachable(_) => {
                        // 标记为不可达，但仍然保留
                    }
                }
                
                let cluster_event = match event {
                    MembershipEvent::MemberJoined(evt) => {
                        ClusterEvent::MemberJoined(MemberJoinedEvent { member: evt.member })
                    }
                    MembershipEvent::MemberLeft(evt) => {
                        ClusterEvent::MemberLeft(MemberLeftEvent { member: evt.member })
                    }
                    MembershipEvent::MemberUnreachable(evt) => {
                        ClusterEvent::MemberUnreachable(MemberUnreachableEvent { member: evt.member })
                    }
                    MembershipEvent::MemberDown(evt) => {
                        ClusterEvent::MemberDown(MemberDownEvent { member: evt.member })
                    }
                };
                
                let _ = event_tx.send(cluster_event).await;
            }
        });
        
        self.monitor_task = Some(task);
        
        // 启动周期性心跳任务
        self.start_heartbeat_task();
        
        // 初始化分区管理器
        if let Some(pm) = &self.partition_manager {
            if let Err(e) = pm.initialize().await {
                tracing::error!("Failed to initialize partition manager: {}", e);
            }
        }
        
        Ok(())
    }
    
    /// 启动心跳任务
    fn start_heartbeat_task(&mut self) {
        // 心跳间隔
        let interval = Duration::from_secs(self.config.heartbeat_interval);
        let provider = self.provider.clone();
        let config = self.config.clone();
        
        let task = tokio::spawn(async move {
            let mut ticker = time::interval(interval);
            
            loop {
                ticker.tick().await;
                
                // 在这里我们可以执行心跳逻辑
                // 例如，重新注册节点以更新成员信息
                if let Err(e) = provider.register(&config).await {
                    tracing::error!("Error sending heartbeat: {}", e);
                }
            }
        });
        
        self.heartbeat_task = Some(task);
    }
    
    /// 停止集群系统
    pub async fn stop(&mut self) -> Result<(), ClusterError> {
        // 注销节点
        self.provider.deregister().await?;
        
        // 停止监控任务
        if let Some(task) = self.monitor_task.take() {
            task.abort();
        }
        
        // 停止心跳任务
        if let Some(task) = self.heartbeat_task.take() {
            task.abort();
        }
        
        Ok(())
    }
    
    /// 获取集群成员列表
    pub async fn get_members(&self) -> Result<Vec<Member>, ClusterError> {
        self.provider.get_members().await
    }
    
    /// 按角色获取集群成员列表
    pub async fn get_members_by_role(&self, role: &str) -> Vec<Member> {
        let roles = self.roles.read().await;
        match roles.get(role) {
            Some(members) => members.clone(),
            None => Vec::new(),
        }
    }
    
    /// 获取虚拟Actor管理器
    pub fn get_grain_manager(&self) -> Option<Arc<grain::GrainManager>> {
        // 假设需要从上下文或其他地方获取GrainManager
        // 这里实现一个简单版本，基于角色直接创建
        
        // 创建GrainManager
        let manager = Arc::new(grain::GrainManager::new(
            Arc::new(self.clone()),
            grain::GrainManagerConfig::default(),
        ));
        
        Some(manager)
    }
    
    /// 在指定节点上远程创建Actor
    pub async fn spawn_remote<A: Actor + 'static>(
        &self,
        address: &str,
        kind: &str,
        name: &str,
        timeout: Duration,
    ) -> AnyhowResult<PID> {
        // 使用远程模块创建远程Actor
        let pid = self.remote.spawn_remote::<A>(address, kind, Some(name), timeout).await?;
        Ok(pid)
    }
    
    /// 发送消息到Actor
    pub async fn send(&self, pid: &PID, msg: impl Any + Send + Sync + 'static) -> AnyhowResult<()> {
        pid.send(msg).await?;
        Ok(())
    }
    
    /// 发送请求消息并等待响应
    pub async fn request<R: Any + Send + Sync + 'static>(
        &self,
        pid: &PID,
        msg: impl Any + Send + Sync + 'static,
        timeout: Duration,
    ) -> AnyhowResult<R> {
        let response = pid.request::<R>(msg, timeout).await?;
        Ok(response)
    }
}

// 实现Clone以支持订阅事件
impl Clone for ClusterSystem {
    fn clone(&self) -> Self {
        Self {
            provider: self.provider.clone(),
            monitor_task: None,
            heartbeat_task: None,
            event_tx: self.event_tx.clone(),
            config: self.config.clone(),
            remote: self.remote.clone(),
            roles: self.roles.clone(),
            partition_manager: None,
        }
    }
} 